package com.yuanfeng.job.service.impl;

import com.yuanfeng.commoms.vo.order.CommissionSettleVO;
import com.yuanfeng.commoms.vo.payment.PayFxAccountVO;
import com.yuanfeng.commoms.vo.payment.PayFxRecordVO;
import com.yuanfeng.job.mapper.OrderCommissionMapper;
import com.yuanfeng.job.mapper.PayFxAccountMapper;
import com.yuanfeng.job.mapper.PayFxRecordMapper;
import com.yuanfeng.job.service.PayFxService;
import com.yuanfeng.job.service.PayFxTaskService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @ClassName PayFxTaskServiceImpl
 * @Description
 * @Author hyq
 * @Date 2023/9/5 13:25
 * @Version 1.0
 **/
@Service
public class PayFxTaskServiceImpl implements PayFxTaskService {

    private static final Logger logger = LoggerFactory.getLogger(PayFxTaskServiceImpl.class);

    @Autowired
    private PayFxAccountMapper payFxAccountMapper;
    @Autowired
    private PayFxRecordMapper payFxRecordMapper;
    @Autowired
    private OrderCommissionMapper orderCommissionMapper;
    @Autowired
    private PayFxService payFxService;
    @Override
    public Boolean vkOrderCommission(List<CommissionSettleVO> commissionSettleList, Integer commissionType) {
        Boolean updateAmountStatus = false;
        ForkJoinPool forkJoinPool = null;
        try {
            forkJoinPool = new ForkJoinPool();
            ForkJoinTask<Boolean> submit = forkJoinPool.submit(new VKOrderCommissionTaskUtils(commissionSettleList,commissionType, 0, commissionSettleList.size() - 1));
            updateAmountStatus = submit.get();
            //阻塞当前线程直到 ForkJoinPool 中所有的任务都执行结束
            forkJoinPool.awaitTermination(2, TimeUnit.SECONDS);
        } catch (Exception e) {
            logger.error(e.getMessage());
        } finally {
            forkJoinPool.shutdown();
        }
        return updateAmountStatus;
    }



    /**
     * 拆分任务
     */
    class VKOrderCommissionTaskUtils extends RecursiveTask<Boolean> {
        /**
         * 每个"小任务"最多只处理的个数
         */
        private static final int MAX = 100;
        List<CommissionSettleVO> taskList;
        Integer commissionType;
        private int start;
        private int end;

        public VKOrderCommissionTaskUtils(List<CommissionSettleVO> taskList,Integer commissionType, int start, int end) {
            this.taskList = taskList;
            this.commissionType = commissionType;
            this.start = start;
            this.end = end;
        }

        @Override
        protected Boolean compute() {
            Boolean flag = false;
            if ((end - start) <= MAX) {
                List<CommissionSettleVO> commissionSettleList = taskList.subList(start, end + 1);

                if (!CollectionUtils.isEmpty(commissionSettleList)) {

                    flag = orderCommission(commissionSettleList,commissionType);

                    return flag;
                }
            } else {
                logger.info( "=====任务分解======");
                // 使用二分法分配任务
                int middle = (start + end) / 2;
                PayFxTaskServiceImpl.VKOrderCommissionTaskUtils left = new PayFxTaskServiceImpl.VKOrderCommissionTaskUtils(taskList,commissionType,start, middle);
                PayFxTaskServiceImpl.VKOrderCommissionTaskUtils right = new PayFxTaskServiceImpl.VKOrderCommissionTaskUtils(taskList,commissionType,middle+1, end);

                // 并行执行两个小任务
                left.fork();
                right.fork();

                Boolean rightFlag = right.join();
                Boolean leftFlag = left.join();
                if(rightFlag && leftFlag){
                    flag = true;
                }
            }
            return flag;
        }
    }

    /**
     * 佣金结算记录
     * @param commissionSettleList
     */
    @Transactional(transactionManager = "payTransactionManager")
    public Boolean orderCommission(List<CommissionSettleVO> commissionSettleList,Integer commissionType) {
        Boolean flag = true;
        try {
            List<String> userIdList = commissionSettleList.stream().map(CommissionSettleVO::getShareUserId).distinct().collect(Collectors.toList());
            // 1.查询所有需要结算用户账单
            List<PayFxAccountVO> payFxAccountList = payFxAccountMapper.queryByUserIdBatch(userIdList);
            Map<String, PayFxAccountVO> payFxAccountMap = payFxAccountList.stream().collect(Collectors.toMap(PayFxAccountVO::getUserId, Function.identity()));

            List<PayFxRecordVO> payFxRecordList = new ArrayList<>();
            List<PayFxAccountVO> addPayAcountList = new ArrayList<>();
            List<PayFxAccountVO> updatePayAcountList = new ArrayList<>();
            // 2.生成结算交易单  更新用户提现金额  用户无账户信息的直接新增
            for (CommissionSettleVO commissionSettle : commissionSettleList) {
                PayFxRecordVO payFxRecord = payFxService.buildFxRecord(commissionSettle, payFxAccountMap.get(commissionSettle.getShareUserId()));
                payFxRecordList.add(payFxRecord);
                // 3.用户账户更新或新增
                if (null == payFxAccountMap.get(commissionSettle.getShareUserId())) {
                    PayFxAccountVO payFxAccount = payFxService.buildFXAccount(commissionSettle);
                    addPayAcountList.add(payFxAccount);
                } else {
                    PayFxAccountVO payFxAccount = new PayFxAccountVO(commissionSettle);
                    updatePayAcountList.add(payFxAccount);
                }
                // 更新当前用户订单装结算单号
                orderCommissionMapper.updateOrderSettleBatch(commissionSettle.getShareUserId(),payFxRecord.getRecordId(),commissionType);
            }
            if (!CollectionUtils.isEmpty(payFxRecordList)) {
                payFxRecordMapper.insertBatch(payFxRecordList);
            }
            if (!CollectionUtils.isEmpty(addPayAcountList)) {
                payFxAccountMapper.insertBatch(addPayAcountList);
            }
            if (!CollectionUtils.isEmpty(updatePayAcountList)) {
                payFxAccountMapper.updateUserAccountBatch(updatePayAcountList);
            }
        } catch (Exception e) {
            flag = false;
            logger.error(e.getMessage());
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
        }
        return flag;
    }
}
